package com.infinitus.autocatlog.service;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.infinitus.autocatlog.domain.KafkaConfig;

public class KafkaSender {

	public static KafkaSender s_sender = null;

	private Producer<String, String> m_producer;

	private KafkaConfig m_configManager;
	
	private String m_topic;

	public static KafkaSender getIntance() {
		synchronized (KafkaSender.class) {
			if (s_sender == null) {
				s_sender = new KafkaSender();
			}
		}
		return s_sender;
	}

	private KafkaSender() {
		try {
			m_configManager = SpringUtil.getBean(KafkaConfig.class);

			m_producer = new KafkaProducer<String, String>(m_configManager.getProperties());
			m_topic = m_configManager.getAutoTopic();
		} catch (Exception e) {
			e.printStackTrace();
			throw new RuntimeException("error when init kakfa producer", e);
		}
	}

	public void send(String str) {
		m_producer.send(new ProducerRecord<String, String>(m_topic, str));
		
//		try {
//	     RecordMetadata result = future.get(5, TimeUnit.SECONDS);
//      } catch (Exception e) {
//	      e.printStackTrace();
//      }
	}

}
